package com.xujian.common.function;

import com.xujian.common.beans.ChannelOrder;
import org.apache.commons.collections.IteratorUtils;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.Iterator;

/**
 * @author star xu
 * @date 2023/1/30 22:28
 * @Description:
 * @Slogan: 致敬大师, 致敬未来的自己
 */
public class MyAggFunction implements WindowFunction<ChannelOrder, Tuple3<String, Integer, Double>, String, TimeWindow> {
    @Override
    public void apply(String s, TimeWindow window, Iterable<ChannelOrder> input, Collector<Tuple3<String, Integer, Double>> out) {
        final int count = IteratorUtils.toList(input.iterator()).size();
        Iterator<ChannelOrder> it = input.iterator();
        double totalAmount = 0;
        while (it.hasNext()) {
            ChannelOrder next = it.next();
            if (next.getPackagePrice() != null) {
                totalAmount += next.getPackagePrice();
            }
        }
        Tuple3<String, Integer, Double> res = new Tuple3<>(s, count, totalAmount);
        out.collect(res);
    }
}
